import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import scala.math.{pow, sqrt}

/**
 * 5、使用鸢尾花数据集实现KMeans算法
 * 假设K = 3
 *    1、读文件data/Iris.csv
 *    2、随机找K个点，作为初始的中心点
 *    3、遍历数据集，计算每一个点与3个中心的距离，距离哪个中心点最近就属于哪个中心点
 *    4、根据新的分类计算新的中心点(求平均值)
 *    5、使用新的中心点开始下一次循环（回到步骤3）
 *
 * 退出循环的条件：
 *    1、指定循环次数
 *    2、所有的中心点几乎不再移动（即所有中心点移动距离的总和小于我们给定的一个常数，如：0.01）
 */

case class Points(species: String, point: Array[Double])

object Subject5 {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)

    //设置分类数为3
    val K = 3
    //设置退出循环条件，即所有中心点移动距离总和的最小值
    val minSum = 0.0001

    //1、读文件data/Iris.csv，得到样本点数据samplePoints
    val sampleRDD: RDD[Points] = sc.textFile("data/Iris.csv").map { line =>
      val fields: Array[String] = line.split(",")
      Points(fields.last, fields.init.tail.map(_.toDouble))
    }

    //2、随机抽取K个点作为初始中心点
    val centerPoints: Array[Array[Double]] = sampleRDD.takeSample(false, K).map(_.point)

    centerPoints.foreach{p =>
      print(s"初始中心点${centerPoints.indexOf(p)}：${p.toBuffer}\t")
      println()
    }
    println("*********************************************************")

    //当前所有中心点移动距离总和，初值设为1.0
    var curSum = 1.0

    //计算两点的欧氏距离
    def getDistance(x: Array[Double], y: Array[Double]) : Double = {
      sqrt(x.zip(y).map(elem => pow(elem._1 - elem._2, 2)).sum)
    }

    //退出循环的条件：所有的中心点几乎不再移动
    while(curSum > minSum){
      //得到每个样本点的分类情况(所属中心点编号, (样本点特征数据, 1))
      val indexRDD: RDD[(Int, (Array[Double], Int))] = sampleRDD.map { sampleP =>
        val dists: Array[Double] = centerPoints.map { centerP =>
          //3、计算样本数据中每个点和中心点的距离
          getDistance(sampleP.point, centerP)
        }
        //(所属中心点编号, (样本点特征数据, 1))
        //所属中心点编号：根据与每个中心点的距离确定所属中心点，距离哪个中心点最近就属于哪个中心点
        //1用于之后计算新中心点
        (dists.indexOf(dists.min), (sampleP.point, 1))
      }

      //4、根据新的分类计算新的中心点(求平均值)
      val newCenterPoints: RDD[(Int, Array[Double])] = indexRDD.reduceByKey { (x, y) =>
        (x._1.zip(y._1).map(elem => elem._1 + elem._2), x._2 + y._2)
      }
        .map { case (index, (point, count)) => (index, point.map(_ / count)) }

      //计算所有中心点移动距离的总和
      val distSum: RDD[Double] = newCenterPoints.map(row => getDistance(row._2, centerPoints(row._1)))
      curSum = distSum.sum

      //5、使用新的中心点，开始下一次循环
      for ((key, value) <- newCenterPoints.collectAsMap()){
        centerPoints(key) = value
      }

      centerPoints.foreach{p =>
        print(s"新中心点${centerPoints.indexOf(p)}：${p.toBuffer}\t")
        println}
      println(s"所有中心点移动距离的总和：$curSum")
      println("*********************************************************")
    }

    centerPoints.foreach{p =>
      print(s"最终中心点${centerPoints.indexOf(p)}：${p.toBuffer}\t")
      println}
    println(s"最终所有中心点移动距离的总和：$curSum")

    sc.stop()
  }
}
